package com.jackstraw.jack_base.datasource.zookeeper.bean;


import com.jackstraw.jack_base.datasource.common.factory.DataSourceBuildFactory;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;


/**
 * Zookeeper工厂类
 */
@Slf4j
public class ZookeeperFactory implements DataSourceBuildFactory<ZookeeperTemplate, ZookeeperPropertiesEntity> {

    @Override
    @SneakyThrows
    public ZookeeperTemplate build(String tenantName, String dataSourceName, ZookeeperPropertiesEntity dataSourceEntity) {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(dataSourceEntity.getBaseSleepTime(), dataSourceEntity.getMaxRetries());
        CuratorFramework curatorFramework = CuratorFrameworkFactory.builder()
                .connectString(dataSourceEntity.getServers())
                .connectionTimeoutMs(dataSourceEntity.getConnectionTimeout())
                .sessionTimeoutMs(dataSourceEntity.getSessionTimeout())
                .retryPolicy(retryPolicy)
                .build();

        curatorFramework.start();

        ZooKeeper zooKeeper = new ZooKeeper(dataSourceEntity.getServers(), dataSourceEntity.getSessionTimeout(), event -> {
            if (Watcher.Event.EventType.None.equals(event.getType())
                    && Watcher.Event.KeeperState.SyncConnected.equals(event.getState())) {
                log.debug("连接已建立");
            }
        });
        ZookeeperTemplate ZookeeperTemplate = new ZookeeperTemplate(curatorFramework, zooKeeper);

        return ZookeeperTemplate;
    }
}
